package io.github.rascaler.dcenter.ws.handler;

import com.alibaba.fastjson.JSON;
import io.github.rascaler.dcenter.annotation.SocketPathMapping;
import io.github.rascaler.dcenter.pojo.Candlestick;
import io.github.rascaler.dcenter.ws.message.KlineSubscribePayload;
import io.github.rascaler.dcenter.ws.message.SocketMessage;
import org.springframework.stereotype.Component;
import org.yeauty.pojo.Session;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @ClassName KlineSubscribeHandler
 * @Description TODO
 * @Author wurenqing
 * @Date 2023-11-28 15:25
 */
@Component
@SocketPathMapping(path="/kline/refresh/subscribe")
public class KlineRefreshSubscribeMessageHandler extends AbstractSocketHandler<KlineSubscribePayload>{

    // 订阅的连接信息
    private Set<Session> subscribers = new HashSet<>();

    private ExecutorService executorService = Executors.newFixedThreadPool(10);


    @Override
    public void execute(Session session, SocketMessage<KlineSubscribePayload> message) {
        // 主题存在的话，加入主题，不存在则创建并加入主题
        subscribers.add(session);
    }

    public void broadcast(Map<String, List<Candlestick>> candlesticks) {
        SocketMessage<KlineSubscribePayload> socketMessage = new SocketMessage<>();
        socketMessage.setId(UUID.randomUUID().toString());
        socketMessage.setPath("/kline/subscribe/broadcast");
        socketMessage.setTimestamp(System.currentTimeMillis());
        KlineSubscribePayload payload = new KlineSubscribePayload();
        payload.setCandlesticks(candlesticks);
        socketMessage.setPayload(payload);
        String json = JSON.toJSONString(socketMessage);
        executorService.submit(() -> {
            for (Session subscriber : subscribers) {
                subscriber.sendText(json);
            }
        });
    }

}
